package top.jidanmian.android.mqttdemo.mqtt

import android.content.Context
import android.util.Log
import org.eclipse.paho.mqttv5.client.IMqttToken
import org.eclipse.paho.mqttv5.client.MqttActionListener
import org.eclipse.paho.mqttv5.client.MqttAsyncClient
import org.eclipse.paho.mqttv5.client.MqttCallback
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence
import org.eclipse.paho.mqttv5.common.MqttException
import org.eclipse.paho.mqttv5.common.MqttMessage
import org.eclipse.paho.mqttv5.common.packet.MqttProperties
import top.jidanmian.android.mqttdemo.R
import java.io.BufferedInputStream
import java.security.KeyStore
import java.security.cert.CertificateFactory
import javax.net.ssl.SSLContext
import javax.net.ssl.TrustManagerFactory

private const val TAG = "MyMqttConn"

class MyMqttConn(private val context: Context) {
    private val serverURI = "ssl://192.168.50.50:8883"
    private val username = "emqx"
    private val password = "public"

    private lateinit var mqttAsyncClient: MqttAsyncClient
    private lateinit var mqttOptions: MqttConnectionOptions

    fun init() {
        mqttAsyncClient =
            MqttAsyncClient(serverURI, System.currentTimeMillis().toString(), MemoryPersistence())

        mqttAsyncClient.setCallback(object : MqttCallback {
            override fun disconnected(disconnectResponse: MqttDisconnectResponse?) {
                Log.d(TAG, "服务器连接断开 - ${disconnectResponse.toString()}")
            }

            override fun mqttErrorOccurred(exception: MqttException?) {
                Log.e(TAG, exception.toString())
            }

            override fun messageArrived(topic: String?, message: MqttMessage?) {
                if (topic == null || message == null) return
                Log.i(TAG, "Topic: $topic, Message: $message")

//                TODO("将收到的信息存入本地数据库中")
            }

            override fun deliveryComplete(token: IMqttToken?) {
                Log.d(TAG, token?.requestMessage.toString())
            }

            override fun connectComplete(reconnect: Boolean, serverURI: String?) {
                Log.i(TAG, "连接服务器：${serverURI.toString()}")
            }

            override fun authPacketArrived(reasonCode: Int, properties: MqttProperties?) {
                Log.i(TAG, properties.toString())
            }
        })

        mqttOptions = MqttConnectionOptions()
        mqttOptions.userName = username
        mqttOptions.password = password.toByteArray()

        val cf = CertificateFactory.getInstance("X.509")
        // app/src/main/res/raw/ca.crt
        val caInput = BufferedInputStream(context.resources.openRawResource(R.raw.ca))
//        // app/src/main/assert/ca.crt
//        val caInput = BufferedInputStream(context.assets.open("ca.crt"))
        val ca = cf.generateCertificate(caInput)
        caInput.close()

        val keyStore = KeyStore.getInstance(KeyStore.getDefaultType())
        keyStore.load(null, null)
        keyStore.setCertificateEntry("ca", ca)

        val tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm())
        tmf.init(keyStore)

        val sslContext = SSLContext.getInstance("SSL", "AndroidOpenSSL")
        sslContext.init(null, tmf.trustManagers, null)

        mqttOptions.socketFactory = sslContext.socketFactory
    }

    fun connect() {
        if (mqttAsyncClient.isConnected) return
        try {
            mqttAsyncClient.connect(mqttOptions, null, object : MqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.d(TAG, "connected")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.d(TAG, "Failed to connect${exception.toString()}")
                }
            })
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

    fun disconnect() {
        if (mqttAsyncClient.isConnected) {
            try {
                mqttAsyncClient.disconnect(null, object : MqttActionListener {
                    override fun onSuccess(asyncActionToken: IMqttToken?) {
                        Log.d(TAG, "Disconnected")
                    }

                    override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                        Log.d(TAG, "Failed to disconnect")
                    }
                })
            } catch (e: MqttException) {
                e.printStackTrace()
            }
        }
    }


    fun publish(topic: String, msg: String, qos: Int = 0, retained: Boolean = false) {
        if (mqttAsyncClient.isConnected) {
            try {
                val message = MqttMessage()
                message.payload = msg.toByteArray()
                message.qos = qos
                message.isRetained = retained
                mqttAsyncClient.publish(topic, message, null, object : MqttActionListener {
                    override fun onSuccess(asyncActionToken: IMqttToken?) {
                        Log.d(TAG, "$msg published to $topic")
                    }

                    override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                        Log.d(TAG, "Failed to publish $msg to $topic")
                    }
                })
            } catch (e: MqttException) {
                e.printStackTrace()
            }
        }
    }

    fun subscribe(topic: String, qos: Int = 0) {
        if (mqttAsyncClient.isConnected) {
            mqttAsyncClient.subscribe(topic, qos, null, object : MqttActionListener {
                override fun onSuccess(asyncActionToken: IMqttToken?) {
                    Log.d(TAG, "订阅成功：$topic")
                }

                override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                    Log.e(TAG, "订阅失败：${exception.toString()}")
                }

            })
        }
    }

    fun unsubscribe(topic: String) {
        if (mqttAsyncClient.isConnected) {
            try {
                mqttAsyncClient.unsubscribe(topic, null, object : MqttActionListener {
                    override fun onSuccess(asyncActionToken: IMqttToken?) {
                        Log.d(TAG, "Unsubscribed to $topic")
                    }

                    override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                        Log.d(TAG, "Failed to unsubscribe $topic")
                    }
                })
            } catch (e: MqttException) {
                e.printStackTrace()
            }
        }
    }
}